package org.apache.lucene.ocean;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.lucene.ocean.util.LongArrayPool;
import org.apache.lucene.ocean.util.SortedListMap;
import org.apache.lucene.search.DocIdSet;
import org.apache.lucene.search.DocIdSetIterator;
import org.apache.lucene.search.Query;
import org.apache.lucene.util.OpenBitSet;

/**
 * Versioned entry.  Performs copy and delete on read.  Deletes are stored in the
 * deletes SortedListMap.  When a value needs to be read, and it is not at the latest 
 * snapshot, a copy is made and the appropriate deletes are made.  
 */
public class BitSetLRUMap {
  private SortedListMap<Long,DocIdSet> deletes = new SortedListMap<Long,DocIdSet>();
  final LinkedHashMap<Query,VersionedEntry> map;
  final ReentrantLock lock = new ReentrantLock();
  final LongArrayPool longPool;
  private final int size;

  public BitSetLRUMap(final int size, final LongArrayPool longPool) {
    this.size = size;
    this.longPool = longPool;
    map = new LinkedHashMap<Query,VersionedEntry>() {
      protected boolean removeEldestEntry(Map.Entry<Query,VersionedEntry> eldest) {
        if (size() > size) {
          VersionedEntry ve = eldest.getValue();
          for (Entry entry : ve.versions.values()) {
            longPool.returnObject(entry.bitSet.getBits());
          }
          return true;
        }
        return false;
      }
    };
  }
  
  /**
   * Remove the bitsets less than the given snapshot id
   * @param id
   */
  public void removeLessThan(long id) {
    lock.lock();
    try {
      Iterator<? extends Map.Entry<Long,DocIdSet>> i1 = deletes.iterator(null, false, id, false);
      while (i1.hasNext()) {
        i1.next();
        i1.remove();
      }
      Iterator<VersionedEntry> i2 = map.values().iterator();
      while (i2.hasNext()) {
        VersionedEntry ve = i2.next();
        Iterator<? extends Map.Entry<Long,Entry>> ive = ve.versions.iterator(null, false, id, false);
        while (ive.hasNext()) {
          Entry entry = ive.next().getValue();
          longPool.returnObject(entry.bitSet.getBits());
          ive.remove();
        }
      }
    } finally {
      lock.unlock();
    }
  }
  
  /**
   * Clone the open bit set.  
   * @param other
   * @return
   */
  // TODO: use the LongArrayPool to reuse long arrays
  public OpenBitSet cloneOpenBitSet(OpenBitSet other) {
    long[] bits = other.getBits();
    long[] cbits = longPool.getObject(bits.length);
    System.arraycopy(bits, 0, cbits, 0, bits.length);
    return new OpenBitSet(cbits, other.getNumWords());
  }
  
  /**
   * For the snapshotid and query given, get the OpenBitSet
   * @param snapshotId
   * @param query
   * @return
   */
  public OpenBitSet get(long snapshotId, Query query) {
    lock.lock();
    try {
      VersionedEntry ve = map.get(query);
      Entry entry = ve.getEntry(snapshotId);
      return entry.bitSet;
    } finally {
      lock.unlock();
    }
  }

  private Entry rectifyEntry(long snapshotId, Entry lastEntry) {
    OpenBitSet cbs = (OpenBitSet) lastEntry.bitSet.clone();
    Iterator<? extends Map.Entry<Long,DocIdSet>> iterator = deletes.iterator(lastEntry.snapshotId, true, snapshotId, true);
    while (iterator.hasNext()) {
      DocIdSet dis = iterator.next().getValue();
      DocIdSetIterator disi = dis.iterator();
      try {
        while (disi.next()) {
          cbs.fastClear(disi.doc());
        }
      } catch (IOException ioe) {
        throw new RuntimeException(ioe);
      }
    }
    return new Entry(snapshotId, cbs);
  }
  
  /**
   * The value stored in the map. It contains multiple versions of the value.
   */
  public class VersionedEntry {
    SortedListMap<Long,Entry> versions = new SortedListMap<Long,Entry>();

    public Entry getEntry(long snapshotId) {
      Entry entry = versions.get(snapshotId);
      if (entry == null) {
        Entry lastEntry = versions.lastValue();
        entry = rectifyEntry(snapshotId, lastEntry);
        versions.put(snapshotId, entry);
      }
      return entry;
    }
  }
  
  /**
   * Entry stored in the VersionedEntry.  Contains the snapshotId and the
   * bitSet.
   */
  public static class Entry {
    public long snapshotId;
    public OpenBitSet bitSet;

    public Entry(long snapshotId, OpenBitSet bitSet) {
      this.snapshotId = snapshotId;
      this.bitSet = bitSet;
    }
  }
}
